package netty.rpc.framework.core.netty.transport.serializer.impl;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import netty.rpc.framework.core.netty.exception.SerializerException;
import netty.rpc.framework.core.netty.transport.protocol.RPCMessage;
import netty.rpc.framework.core.netty.transport.protocol.RPCRequest;
import netty.rpc.framework.core.netty.transport.protocol.RPCResponse;
import netty.rpc.framework.core.netty.transport.serializer.Serializer;

import java.io.*;

/**
 * @author guiyiqiao
 * @date 2020/12/4
 */
@Slf4j
public class KryoSerializer implements Serializer {

    /**
     * kryo非线程安全类，使用ThreadLocal保证线程安全
     */

    ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(
            ()->{
                Kryo kryo = new Kryo();
                //kryo.register(RPCMessage.class);
                return kryo;
            }
    );

    @Override
    public byte[] serialize(@NonNull Object obj) {
        try(
                ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                Output output = new Output(byteArrayOutputStream);
                ) {
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output,obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (IOException e) {
            throw new SerializerException();
        }
    }

    @Override
    public <T> T deserialize(@NonNull byte[] bytes,@NonNull Class<T> clazz) {


        try (
            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
            Input input = new Input(byteArrayInputStream);
        ) {
            Kryo kryo = kryoThreadLocal.get();
            Object message = kryo.readObject(input, clazz);
            log.info("序列化了一个 msg = {}",message.toString());
            kryoThreadLocal.remove();

            return clazz.cast(message);
        } catch (IOException e) {
            throw new SerializerException();
        }
    }
}
